package top.wshape1.rpc.protocol;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;
import lombok.extern.slf4j.Slf4j;
import top.wshape1.rpc.compress.CompressorFactory;
import top.wshape1.rpc.config.JJRpcConfig;
import top.wshape1.rpc.message.RpcMessage;
import top.wshape1.rpc.message.serializer.SerializerFactory;

import java.util.List;

/**
 * @author Wshape1
 * @version 1.0
 * @description 消息编解码器
 * @since 2024-02-01
 */

@Slf4j
@ChannelHandler.Sharable
public class RpcMessageCodec extends MessageToMessageCodec<ByteBuf, RpcMessage> {

    private static final int MAGIC_NUMBER = 0x77336666;
    private static final int VERSION = 1;

    /*
     * 数据格式包含：魔数 + 版本 + 消息类型 + 请求序列号 + 是否压缩 + 无用占位 + 长度 + 消息本体
     *      byte：  4    1        1        4         1        1        4
     *
     */

    private final boolean isCompress;

    public RpcMessageCodec() {
        isCompress = JJRpcConfig.CONFIG.getCompress().isEnabled();
    }

    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, RpcMessage rpcMessage, List<Object> list) {
        log.debug("RpcMessageCodec开始编码消息，消息类型：{}，消息内容：{}", rpcMessage.getClass(), rpcMessage);

        ByteBuf buffer = channelHandlerContext.alloc().buffer();

        // 4 byte 魔数
        buffer.writeInt(MAGIC_NUMBER);

        // 1 byte 协议版本
        buffer.writeByte(VERSION);

        // 1 byte 消息类型
        buffer.writeByte(rpcMessage.getMessageType());

        // 4 byte 请求序列号
        buffer.writeInt(rpcMessage.getSequenceId());

        // 1 byte 是否压缩
        buffer.writeByte(isCompress ? 1 : 0);

        // 1 byte 无用占位
        buffer.writeByte(0);

        // 序列化消息
        byte[] bytes;
        try {
            bytes = SerializerFactory.getSerializer().serialize(rpcMessage);
            if (isCompress)
                bytes = CompressorFactory.getCompressor().compress(bytes);
        } catch (Exception e) {
            log.error("Encode Message failed, error: {}", e.getMessage(), e);
            throw new RuntimeException(e);
        }

        // 4 byte 消息长度
        buffer.writeInt(bytes.length);

        // 消息本体
        buffer.writeBytes(bytes);

        list.add(buffer);
        log.debug("RpcMessageCodec编码消息成功");

    }

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) {
        log.debug("RpcMessageCodec开始解码消息");
        int magic = byteBuf.readInt();
        if (magic != MAGIC_NUMBER) {
            log.warn("Decode failed, error magic: int({})", magic);
            return;
        }

        byte version = byteBuf.readByte();
        if (version != VERSION) {
            log.warn("Decode failed, error version: {}", version);
            return;
        }

        byte msgType = byteBuf.readByte();
        byteBuf.readInt(); // 序列号
        byte compress = byteBuf.readByte();
        byteBuf.readByte();
        int length = byteBuf.readInt();

        // 读取消息本体
        byte[] bytes = new byte[length];
        byteBuf.readBytes(bytes, 0, length);

        // 反序列化消息
        RpcMessage message;
        try {
            if (compress == 1) {
                bytes = CompressorFactory.getCompressor().decompress(bytes);
            }
            message = SerializerFactory
                    .getSerializer()
                    .deserialize(bytes, RpcMessage.getMessageType(msgType));
        } catch (Exception e) {
            log.error("Decode failed, error: {}", e.getMessage(), e);
            throw new RuntimeException(e);
        }

        if (message != null) {
            list.add(message);
            log.debug("RpcMessageCodec解码消息成功，消息类型：{}，消息内容：{}", message.getClass(), message);
        }
    }
}
